﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;
using System.Diagnostics;
using System.Runtime.Serialization.Formatters.Binary;
using System.IO;
using System.Reflection;
using Go;

namespace GoRpc
{
    using MethodsMap = Dictionary<string, Func<async_result_wrap<object, Rpc.ReplyState>, object[], Task>>;

    public class Rpc
    {
        enum DataType
        {
            Request,
            Reply,
        }

        [Serializable]
        struct Request
        {
            public string name;
            public object[] args;
        }

        [Serializable]
        struct Reply
        {
            public ReplyState state;
            public string message;
            public object result;
        }

        public enum ReplyState
        {
            Success,
            Overtime,
            Stopped,
            Exception,
            NotFound,
            Fail,
        }

        public class Methods
        {
            struct BindObjInfo
            {
                public bool isTask;
                public MethodInfo method;
                public MethodInfo valueTask;
                public MethodInfo getResult;
            }

            protected bool _noBind = false;
            internal MethodsMap _methodsMap = new MethodsMap();

            public void Bind(string name, Func<Task> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => handler());
            }

            public void Bind<T1>(string name, Func<T1, Task> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => handler((T1)args[0]));
            }

            public void Bind<T1, T2>(string name, Func<T1, T2, Task> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => handler((T1)args[0], (T2)args[1]));
            }

            public void Bind<T1, T2, T3>(string name, Func<T1, T2, T3, Task> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => handler((T1)args[0], (T2)args[1], (T3)args[2]));
            }

            private async Task bind_wait<R>(async_result_wrap<object, ReplyState> res, Task<R> task)
            {
                res.value1 = await task;
            }

            public void Bind<R>(string name, Func<Task<R>> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, delegate (async_result_wrap<object, ReplyState> res, object[] args)
                {
                    Task<R> task = handler();
                    if (!task.IsCompleted)
                    {
                        return bind_wait(res, task);
                    }
                    res.value1 = task.GetAwaiter().GetResult();
                    return generator.non_async();
                });
            }

            public void Bind<R, T1>(string name, Func<T1, Task<R>> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, delegate (async_result_wrap<object, ReplyState> res, object[] args)
                {
                    Task<R> task = handler((T1)args[0]);
                    if (!task.IsCompleted)
                    {
                        return bind_wait(res, task);
                    }
                    res.value1 = task.GetAwaiter().GetResult();
                    return generator.non_async();
                });
            }

            public void Bind<R, T1, T2>(string name, Func<T1, T2, Task<R>> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, delegate (async_result_wrap<object, ReplyState> res, object[] args)
                {
                    Task<R> task = handler((T1)args[0], (T2)args[1]);
                    if (!task.IsCompleted)
                    {
                        return bind_wait(res, task);
                    }
                    res.value1 = task.GetAwaiter().GetResult();
                    return generator.non_async();
                });
            }

            public void Bind<R, T1, T2, T3>(string name, Func<T1, T2, T3, Task<R>> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, delegate (async_result_wrap<object, ReplyState> res, object[] args)
                {
                    Task<R> task = handler((T1)args[0], (T2)args[1], (T3)args[2]);
                    if (!task.IsCompleted)
                    {
                        return bind_wait(res, task);
                    }
                    res.value1 = task.GetAwaiter().GetResult();
                    return generator.non_async();
                });
            }

            private async Task bind_wait<R>(async_result_wrap<object, ReplyState> res, ValueTask<R> task)
            {
                res.value1 = await task;
            }

            public void Bind<R>(string name, Func<ValueTask<R>> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, delegate (async_result_wrap<object, ReplyState> res, object[] args)
                {
                    ValueTask<R> task = handler();
                    if (!task.IsCompleted)
                    {
                        return bind_wait(res, task);
                    }
                    res.value1 = task.GetAwaiter().GetResult();
                    return generator.non_async();
                });
            }

            public void Bind<R, T1>(string name, Func<T1, ValueTask<R>> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, delegate (async_result_wrap<object, ReplyState> res, object[] args)
                {
                    ValueTask<R> task = handler((T1)args[0]);
                    if (!task.IsCompleted)
                    {
                        return bind_wait(res, task);
                    }
                    res.value1 = task.GetAwaiter().GetResult();
                    return generator.non_async();
                });
            }

            public void Bind<R, T1, T2>(string name, Func<T1, T2, ValueTask<R>> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, delegate (async_result_wrap<object, ReplyState> res, object[] args)
                {
                    ValueTask<R> task = handler((T1)args[0], (T2)args[1]);
                    if (!task.IsCompleted)
                    {
                        return bind_wait(res, task);
                    }
                    res.value1 = task.GetAwaiter().GetResult();
                    return generator.non_async();
                });
            }

            public void Bind<R, T1, T2, T3>(string name, Func<T1, T2, T3, ValueTask<R>> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, delegate (async_result_wrap<object, ReplyState> res, object[] args)
                {
                    ValueTask<R> task = handler((T1)args[0], (T2)args[1], (T3)args[2]);
                    if (!task.IsCompleted)
                    {
                        return bind_wait(res, task);
                    }
                    res.value1 = task.GetAwaiter().GetResult();
                    return generator.non_async();
                });
            }

            public void Bind(string name, Action handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => { handler(); return generator.non_async(); });
            }

            public void Bind<T1>(string name, Action<T1> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => { handler((T1)args[0]); return generator.non_async(); });
            }

            public void Bind<T1, T2>(string name, Action<T1, T2> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => { handler((T1)args[0], (T2)args[1]); return generator.non_async(); });
            }

            public void Bind<T1, T2, T3>(string name, Action<T1, T2, T3> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => { handler((T1)args[0], (T2)args[1], (T3)args[2]); return generator.non_async(); });
            }

            public void Bind<R>(string name, Func<R> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => { res.value1 = handler(); return generator.non_async(); });
            }

            public void Bind<R, T1>(string name, Func<T1, R> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => { res.value1 = handler((T1)args[0]); return generator.non_async(); });
            }

            public void Bind<R, T1, T2>(string name, Func<T1, T2, R> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => { res.value1 = handler((T1)args[0], (T2)args[1]); return generator.non_async(); });
            }

            public void Bind<R, T1, T2, T3>(string name, Func<T1, T2, T3, R> handler)
            {
                Debug.Assert(!_noBind, "不正确的 Bind 操作，服务已经启动");
                _methodsMap.Add(name, (async_result_wrap<object, ReplyState> res, object[] args) => { res.value1 = handler((T1)args[0], (T2)args[1], (T3)args[2]); return generator.non_async(); });
            }

            public void BindObj(string name, object obj)
            {
                MethodInfo[] methods = obj.GetType().GetMethods(BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public | BindingFlags.Static);
                Dictionary<string, LinkedList<BindObjInfo>> overrideMethod = new Dictionary<string, LinkedList<BindObjInfo>>();
                for (int i = 0; i < methods.Length; i++)
                {
                    BindObjInfo methodInfo = new BindObjInfo { method = methods[i] };
                    Type returnType = methodInfo.method.ReturnType;
                    if ((methodInfo.isTask = ("Task`1" == returnType.Name || "Task" == returnType.Name)) &&
                        "System.Threading.Tasks.Task`1[System.Threading.Tasks.VoidTaskResult]" != returnType.ToString())
                    {
                        methodInfo.getResult = returnType.GetMethod("get_Result", BindingFlags.Instance | BindingFlags.Public);
                    }
                    else if ("ValueTask`1" == returnType.Name)
                    {
                        methodInfo.valueTask = returnType.GetMethod("AsTask", BindingFlags.Instance | BindingFlags.Public);
                        methodInfo.getResult = returnType.GetMethod("get_Result", BindingFlags.Instance | BindingFlags.Public);
                    }
                    LinkedList<BindObjInfo> dupName;
                    string methodName = methodInfo.method.Name;
                    if (!overrideMethod.TryGetValue(methodName, out dupName))
                    {
                        dupName = new LinkedList<BindObjInfo>();
                        overrideMethod.Add(methodName, dupName);
                        _methodsMap.Add(string.Format("{0}.{1}", name, methodName), async delegate (async_result_wrap<object, ReplyState> res, object[] args)
                        {
                            if (1 == dupName.Count)
                            {
                                BindObjInfo invokeMethod = dupName.First.Value;
                                if (invokeMethod.isTask)
                                {
                                    Task result;
                                    await (result = (Task)invokeMethod.method.Invoke(obj, args));
                                    res.value1 = null == invokeMethod.getResult ? null : invokeMethod.getResult.Invoke(result, null);
                                }
                                else if (null != invokeMethod.valueTask)
                                {
                                    object result = invokeMethod.method.Invoke(obj, args);
                                    Task task = (Task)invokeMethod.valueTask.Invoke(result, null);
                                    if (null != task) await task;
                                    res.value1 = invokeMethod.getResult.Invoke(result, null);
                                }
                                else
                                {
                                    res.value1 = await generator.send_task(() => invokeMethod.method.Invoke(obj, args));
                                }
                                return;
                            }
                            for (LinkedListNode<BindObjInfo> it = dupName.First; null != it; it = it.Next)
                            {
                                ParameterInfo[] parameters = it.Value.method.GetParameters();
                                if (parameters.Length == args.Length)
                                {
                                    int j = 0;
                                    for (; j < args.Length && parameters[j].ParameterType == args[j].GetType(); j++) { }
                                    if (j == args.Length)
                                    {
                                        BindObjInfo invokeMethod = it.Value;
                                        if (invokeMethod.isTask)
                                        {
                                            Task result;
                                            await (result = (Task)invokeMethod.method.Invoke(obj, args));
                                            res.value1 = null == invokeMethod.getResult ? null : invokeMethod.getResult.Invoke(result, null);
                                        }
                                        else if (null != invokeMethod.valueTask)
                                        {
                                            object result = invokeMethod.method.Invoke(obj, args);
                                            Task task = (Task)invokeMethod.valueTask.Invoke(result, null);
                                            if (null != task) await task;
                                            res.value1 = invokeMethod.getResult.Invoke(result, null);
                                        }
                                        else
                                        {
                                            res.value1 = await generator.send_task(() => invokeMethod.method.Invoke(obj, args));
                                        }
                                        return;
                                    }
                                }
                            }
                            res.value2 = ReplyState.Exception;
                        });
                    }
                    dupName.AddLast(methodInfo);
                }

                FieldInfo[] fields = obj.GetType().GetFields(BindingFlags.Instance | BindingFlags.NonPublic | BindingFlags.Public | BindingFlags.Static);
                for (int i = 0; i < fields.Length; i++)
                {
                    try
                    {
                        FieldInfo field = fields[i];
                        _methodsMap.Add(string.Format("{0}.get@{1}", name, field.Name), delegate (async_result_wrap<object, ReplyState> res, object[] args)
                        {
                            res.value1 = field.GetValue(obj);
                            return generator.non_async();
                        });
                        _methodsMap.Add(string.Format("{0}.set@{1}", name, field.Name), delegate (async_result_wrap<object, ReplyState> res, object[] args)
                        {
                            field.SetValue(obj, args[0]);
                            return generator.non_async();
                        });
                    }
                    catch (System.ArgumentException) { }
                }
            }

            public void bind_block(string name, Action handler)
            {
                Bind(name, () => generator.send_task(handler));
            }

            public void bind_block<T1>(string name, Action<T1> handler)
            {
                Bind(name, (T1 p1) => generator.send_task(() => handler(p1)));
            }

            public void bind_block<T1, T2>(string name, Action<T1, T2> handler)
            {
                Bind(name, (T1 p1, T2 p2) => generator.send_task(() => handler(p1, p2)));
            }

            public void bind_block<T1, T2, T3>(string name, Action<T1, T2, T3> handler)
            {
                Bind(name, (T1 p1, T2 p2, T3 p3) => generator.send_task(() => handler(p1, p2, p3)));
            }

            public void bind_block<R>(string name, Func<R> handler)
            {
                Bind(name, () => generator.send_task(handler));
            }

            public void bind_block<R, T1>(string name, Func<T1, R> handler)
            {
                Bind(name, (T1 p1) => generator.send_task(() => handler(p1)));
            }

            public void bind_block<R, T1, T2>(string name, Func<T1, T2, R> handler)
            {
                Bind(name, (T1 p1, T2 p2) => generator.send_task(() => handler(p1, p2)));
            }

            public void bind_block<R, T1, T2, T3>(string name, Func<T1, T2, T3, R> handler)
            {
                Bind(name, (T1 p1, T2 p2, T3 p3) => generator.send_task(() => handler(p1, p2, p3)));
            }
        }

        public class RpcException : Exception
        {
            public ReplyState code;

            public RpcException(ReplyState st)
            {
                code = st;
            }
        }

        public struct Result
        {
            public ReplyState state;
            public string message;

            public void Check()
            {
                if (ReplyState.Success != state)
                {
                    throw new RpcException(state);
                }
            }
        }

        public struct Result<R>
        {
            public ReplyState state;
            public string message;
            public R result;

            public static implicit operator R(Result<R> rval)
            {
                rval.Check();
                return rval.result;
            }

            public void Check()
            {
                if (ReplyState.Success != state)
                {
                    throw new RpcException(state);
                }
            }
        }

        public class Session
        {
            socket _socket;
            Methods _methods;
            go_mutex _mutex;
            csp_chan<Reply, tuple<int, Request>> _rpcChan;

            public Session(socket socket, Methods methods)
            {
                _socket = socket;
                _methods = methods;
                shared_strand strand = new shared_strand();
                _mutex = new go_mutex(strand);
                _rpcChan = new csp_chan<Reply, tuple<int, Request>>(strand);
            }

            static private int Read32(byte[] buff, int offset)
            {
                int result = 0;
                for (int i = 0; i < 4 && i + offset < buff.Length; i++)
                {
                    result |= buff[i + offset] << (i * 8);
                }
                return result;
            }

            static private long Read64(byte[] buff, int offset)
            {
                long result = 0;
                for (int i = 0; i < 8 && i + offset < buff.Length; i++)
                {
                    result |= ((long)buff[i + offset]) << (i * 8);
                }
                return result;
            }

            static private void Write32(int val, byte[] buff, int offset)
            {
                for (int i = 0; i < 4 && i + offset < buff.Length; i++)
                {
                    buff[i + offset] = (byte)(val >> (i * 8));
                }
            }

            static private void Write64(long val, byte[] buff, int offset)
            {
                for (int i = 0; i < 8 && i + offset < buff.Length; i++)
                {
                    buff[i + offset] = (byte)(val >> (i * 8));
                }
            }

            public async Task<Result> Call(int ms, string name, params object[] args)
            {
                csp_invoke_wrap<Reply> result = await generator.csp_invoke(_rpcChan, new tuple<int, Request>(ms, new Request { name = name, args = args }));
                if (chan_state.ok != result.state)
                {
                    return new Result { state = ReplyState.Fail };
                }
                else if (ReplyState.Success != result.result.state)
                {
                    return new Result { state = result.result.state, message = result.result.message };
                }
                return new Result { state = result.result.state };
            }

            public Task<Result> Call(string name, params object[] args)
            {
                return Call(-1, name, args);
            }

            public async Task<Result<R>> Call<R>(int ms, string name, params object[] args)
            {
                csp_invoke_wrap<Reply> result = await generator.csp_invoke(_rpcChan, new tuple<int, Request>(ms, new Request { name = name, args = args }));
                if (chan_state.ok != result.state)
                {
                    return new Result<R> { state = ReplyState.Fail };
                }
                else if (ReplyState.Success != result.result.state)
                {
                    return new Result<R> { state = result.result.state, message = result.result.message };
                }
                return new Result<R> { state = result.result.state, result = (R)result.result.result };
            }

            public Task<Result<R>> Call<R>(string name, params object[] args)
            {
                return Call<R>(-1, name, args);
            }

            public void Close()
            {
                _rpcChan.close();
            }

            public async Task Run()
            {
                generator.lock_suspend();
                await generator.mutex_lock(_mutex);
                if (null == _socket)
                {
                    await generator.mutex_unlock(_mutex);
                    await generator.unlock_suspend();
                    return;
                }
                generator.children ioChild = new generator.children();
                var replyMap = new Dictionary<long, tuple<async_timer, csp_wait_wrap<Reply, tuple<int, Request>>>>();
                try
                {
                    chan<tuple<long, Request>> recvRequestChan = chan<tuple<long, Request>>.make(-1);
                    chan<tuple<long, Reply>> recvReplyChan = chan<tuple<long, Reply>>.make(-1);
                    chan<tuple<long, Request>> sendRequestChan = chan<tuple<long, Request>>.make(-1);
                    chan<tuple<long, Reply>> sendReplyChan = chan<tuple<long, Reply>>.make(-1);
                    ioChild.go(async delegate ()
                    {
                        byte[] recvHead = new byte[13];
                        byte[] recvBuff = new byte[1024 * 1024];
                        async_timer timer = new async_timer();
                        timer.timeout(10000, () => _rpcChan.close());
                        BinaryFormatter deserializer = new BinaryFormatter();
                        MemoryStream recvStream = new MemoryStream(recvBuff, 0, recvBuff.Length);
                        try
                        {
                            while (true)
                            {
                                if (!await _socket.read(recvHead))
                                {
                                    goto exit;
                                }
                                timer.restart();
                                int len = Read32(recvHead, 0);
                                if (len <= 0)
                                {
                                    continue;
                                }
                                if (len > recvBuff.Length)
                                {
                                    goto exit;
                                }
                                if (!await _socket.read(new ArraySegment<byte>(recvBuff, 0, len)))
                                {
                                    goto exit;
                                }
                                long id = Read64(recvHead, 4);
                                DataType type = (DataType)recvHead[12];
                                recvStream.Position = 0;
                                switch (type)
                                {
                                    case DataType.Request:
                                        await recvRequestChan.send(tuple.make(id, await generator.send_task(() => (Request)deserializer.Deserialize(recvStream))));
                                        break;
                                    case DataType.Reply:
                                        await recvReplyChan.send(tuple.make(id, await generator.send_task(() => (Reply)deserializer.Deserialize(recvStream))));
                                        break;
                                }
                            }
                            exit:;
                        }
                        catch (generator.stop_exception) { throw; }
                        catch (Exception) { }
                        finally
                        {
                            timer.cancel();
                            _rpcChan.close();
                        }
                    });
                    ioChild.go(async delegate ()
                    {
                        byte[] sendHead = new byte[13];
                        byte[] sendBuff = new byte[1024 * 1024];
                        async_timer timer = new async_timer();
                        chan<void_type> keepChan = chan<void_type>.make(1);
                        timer.interval(5000, keepChan.wrap_try_default());
                        BinaryFormatter serializer = new BinaryFormatter();
                        MemoryStream sendStream = new MemoryStream(sendBuff, 0, sendBuff.Length);
                        try
                        {
                            await generator.select().case_receive(sendReplyChan, async delegate (tuple<long, Reply> reply)
                            {
                                sendStream.Position = 0;
                                await generator.send_task(() => serializer.Serialize(sendStream, reply.value2));
                                Write32((int)sendStream.Position, sendHead, 0);
                                Write64(reply.value1, sendHead, 4);
                                sendHead[12] = (byte)DataType.Reply;
                                if (!await _socket.write(sendHead))
                                {
                                    generator.select_chans.stop_all();
                                }
                                if (!await _socket.write(new ArraySegment<byte>(sendBuff, 0, (int)sendStream.Position)))
                                {
                                    generator.select_chans.stop_all();
                                }
                            }).case_receive(sendRequestChan, async delegate (tuple<long, Request> request)
                            {
                                sendStream.Position = 0;
                                await generator.send_task(() => serializer.Serialize(sendStream, request.value2));
                                Write32((int)sendStream.Position, sendHead, 0);
                                Write64(request.value1, sendHead, 4);
                                sendHead[12] = (byte)DataType.Request;
                                if (!await _socket.write(sendHead))
                                {
                                    generator.select_chans.stop_all();
                                }
                                if (!await _socket.write(new ArraySegment<byte>(sendBuff, 0, (int)sendStream.Position)))
                                {
                                    generator.select_chans.stop_all();
                                }
                            }).case_receive(keepChan, async delegate ()
                            {
                                Write32(0, sendHead, 0);
                                Write64(0, sendHead, 4);
                                sendHead[12] = 0;
                                if (!await _socket.write(sendHead))
                                {
                                    generator.select_chans.stop_all();
                                }
                            }).loop();
                        }
                        catch (generator.stop_exception) { throw; }
                        catch (Exception) { }
                        finally
                        {
                            timer.cancel();
                            _rpcChan.close();
                        }
                    });
                    ioChild.go(async delegate ()
                    {
                        generator.children callChild = new generator.children();
                        while (true)
                        {
                            tuple<long, Request> request = await recvRequestChan.receive();
                            callChild.go(async delegate ()
                            {
                                Reply reply = new Reply { };
                                try
                                {
                                    Func<async_result_wrap<object, ReplyState>, object[], Task> handler;
                                    if (_methods._methodsMap.TryGetValue(request.value2.name, out handler))
                                    {
                                        async_result_wrap<object, ReplyState> result = new async_result_wrap<object, ReplyState> { value2 = ReplyState.Success };
                                        try
                                        {
                                            await handler(result, request.value2.args);
                                            reply.result = result.value1;
                                            reply.state = result.value2;
                                        }
                                        catch (generator.stop_exception)
                                        {
                                            reply.state = ReplyState.Stopped;
                                            throw;
                                        }
                                        catch (Exception ec)
                                        {
                                            reply.state = ReplyState.Exception;
                                            reply.message = ec.Message;
                                        }
                                    }
                                    else
                                    {
                                        reply.state = ReplyState.NotFound;
                                    }
                                }
                                finally
                                {
                                    await generator.chan_send(sendReplyChan, tuple.make(request.value1, reply));
                                }
                            });
                        }
                    });
                    ioChild.go(async delegate ()
                    {
                        while (true)
                        {
                            tuple<long, Reply> reply = await recvReplyChan.receive();
                            if (replyMap.TryGetValue(reply.value1, out var repSign))
                            {
                                replyMap.Remove(reply.value1);
                                repSign.value1?.cancel();
                                repSign.value2.complete(reply.value2);
                            }
                        }
                    });
                    long idCount = 0;
                    while (true)
                    {
                        long id = ++idCount;
                        var request = await _rpcChan.wait();
                        if (request.state != chan_state.ok)
                        {
                            break;
                        }
                        if (request.msg.value1 >= 0)
                        {
                            async_timer timer = new async_timer();
                            replyMap.Add(id, tuple.make(timer, request));
                            timer.timeout(request.msg.value1, delegate ()
                            {
                                replyMap.Remove(id);
                                request.fail();
                            });
                        }
                        else
                        {
                            replyMap.Add(id, tuple.make((async_timer)null, request));
                        }
                        await sendRequestChan.send(tuple.make(id, request.msg.value2));
                    }
                }
                finally
                {
                    generator.lock_stop();
                    await ioChild.stop();
                    _rpcChan.close();
                    _socket.close();
                    _socket = null;
                    foreach (var item in replyMap)
                    {
                        item.Value.value1?.cancel();
                        item.Value.value2.fail();
                    }
                    generator.unlock_stop();
                    await generator.mutex_unlock(_mutex);
                    await generator.unlock_suspend();
                }
            }
        }
    }
}
